package cn.gs.exchanger;

// @author gs
import cn.gs.exchanger.ExchangerCommand.SystemCommand;
import cn.gs.util.net.ReConnectSocket;
import cn.gs.util.set.Pair;
import java.io.IOException;
import java.net.UnknownHostException;
import java.util.Map.Entry;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

public class ExchangerClient extends ReConnectSocket {

    public static final int PORT = 82;
    public static int TIMEOUT = 30000;

    public static interface IService {

        List<String> process(List<String> args) throws ExchangerException;
    }

    protected List<String> dealAnswer(Pair<List<String>, ExchangerException> ans) throws ExchangerException {
        if (ans.getSecond() != null) {
            throw ans.getSecond();
        }
        return ans.getFirst();
    }

    /**
     * Standard service register
     *
     * @param name
     * @param service
     * @throws ExchangerException
     */
    public void register(String name, IService service, boolean unique) throws ExchangerException {
        services.put(name, service);
        String server = unique ? SystemCommand.UNIQUE.getServer() : SystemCommand.REGISTER.getServer();
        Pair<List<String>, ExchangerException> ans = ask(server, name);
        dealAnswer(ans);
        Logger.getLogger(ExchangerClient.class.getName()).log(Level.INFO, "{0}_SERVICE_{1}", new Object[]{unique ? "UNI" : "REG", name});
    }

    /**
     * Standard service remover
     *
     * @param name
     * @throws ExchangerException
     */
    public void delete(String name) throws ExchangerException {
        Pair<List<String>, ExchangerException> ans = ask(SystemCommand.DELETE.getServer(), name);
        dealAnswer(ans);
        services.remove(name);
        Logger.getLogger(ExchangerClient.class.getName()).log(Level.INFO, "DELETED_SERVICE_{1}", new Object[]{name});
    }

    /**
     * Standard ask processer, only using of String args and return String
     *
     * @param name ServerName
     * @param args Arguments
     * @return Pair of answer and whether it is error or not
     */
    public Pair<List<String>, ExchangerException> ask(String name, List<String> args) {
        ExchangerCommand req = new ExchangerCommand(UUID.randomUUID().toString(), name, args);
        Pair<ExchangerCommand, ExchangerCommand> lock = null;
        try {
            lock = new Pair<ExchangerCommand, ExchangerCommand>(req, null);
            synchronized (lock) {
                lockers.put(req.getNumber(), lock);
                askai.incrementAndGet();
                req.sendTo(this);
                Logger.getLogger(ExchangerClient.class.getName()).log(Level.FINEST, "Sent_{0}", req.getNumber());
                //waiting for response
                lock.wait(TIMEOUT);
                Logger.getLogger(ExchangerClient.class.getName()).log(Level.FINEST, "Await_{0}", req.getNumber());
                if (lock == null || lock.getSecond() == null) {
                    return new Pair<List<String>, ExchangerException>(null, ExchangerException.Type.SERVER_NOT_RESPONSE.create(req.toString(), null));
                }
            }
            lock = lockers.remove(req.getNumber());
        } catch (InterruptedException ex) {
            return new Pair<List<String>, ExchangerException>(null, ExchangerException.Type.SERVER_NOT_RESPONSE.create(req.getNumber(), ex));
        } catch (IOException ex) {
            return new Pair<List<String>, ExchangerException>(null, ExchangerException.Type.IO_ERROR.create(req.getNumber(), ex));
        }
        req = lock.getSecond();
        if (req.match(ExchangerCommand.SystemCommand.ANSWER)) {
            return new Pair<List<String>, ExchangerException>(req.getArgsList(), null);
        } else if (req.match(ExchangerCommand.SystemCommand.ERROR)) {
            return new Pair<List<String>, ExchangerException>(null, new ExchangerException(req.getArgsList()));
        } else {
            return new Pair<List<String>, ExchangerException>(null, ExchangerException.Type.UNKNOWN_EXCEPTION.create(req.getNumber() + ":Response Command " + req.getServer(), null));
        }
    }

    public Pair<List<String>, ExchangerException> ask(String name, String... args) {
        return ask(name, Arrays.asList(args));
    }

    public List<Integer> serviceProviderCount(String... services) throws ExchangerException {
        List<String> ls = dealAnswer(ask(ExchangerCommand.SystemCommand.SERVICE_PROVIDER_COUNT.getServer(), Arrays.asList(services)));
        List<Integer> is = new ArrayList<Integer>(ls.size());
        for (int i = 0; i < ls.size(); ++i) {
            try {
                is.add(Integer.valueOf(ls.get(i)));
            } catch (NumberFormatException ex) {
                is.add(-1);
            }
        }
        return is;
    }

    public List<String> allService() throws ExchangerException {
        return dealAnswer(ask(ExchangerCommand.SystemCommand.ALL_SERVICE.getServer(), new LinkedList<String>()));
    }

    @Override
    protected void onDisconnect(Throwable th) {
        AfterReConnect a = new AfterReConnect();
        a.objs = lockers.values();
        pool.execute(a);
        //objs.removeAll(objs);
        super.onDisconnect(th);
    }

    class AfterReConnect implements Runnable {

        Collection<Pair<ExchangerCommand, ExchangerCommand>> objs;

        @Override
        public void run() {
            Thread.currentThread().setName("AfterReConnect");
            //Re-send Command Request
            for (Pair<ExchangerCommand, ExchangerCommand> lock : objs) {
                synchronized (lock) {
                    try {
                        lock.getFirst().sendTo(ExchangerClient.this);
                        lockers.put(lock.getFirst().getNumber(), lock);
                    } catch (IOException ex) {
                        Logger.getLogger(ExchangerClient.class.getName()).log(Level.SEVERE, null, ex);
                    }
                }
            }
            //Re-Register
            Map<String, IService> ser = new HashMap<String, IService>(services);
            services.clear();
            for (Entry<String, IService> e : ser.entrySet()) {
                try {
                    register(e.getKey(), e.getValue(), false);
                } catch (ExchangerException ex) {
                    Logger.getLogger(ExchangerClient.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    }

    private class Listener extends Thread {

        Listener() {
            super("ExchangerClient");
        }

        public void run() {
            Logger.getLogger(ExchangerClient.class.getName()).log(Level.INFO, "ExchangerClient_Start");
            while (!Thread.interrupted()) {
                String msg = readLine();
                ExchangerCommand cmd = new ExchangerCommand(msg);
                if (cmd.isSystemCommand()) {
                    if (cmd.match(ExchangerCommand.SystemCommand.ANSWER) || cmd.match(ExchangerCommand.SystemCommand.ERROR)) {
                        Pair<ExchangerCommand, ExchangerCommand> lock = lockers.get(cmd.getNumber());
                        if (lock != null) {
                            synchronized (lock) {
                                lock.setSecond(cmd);
                                lock.notify();
                            }
                        }
                    }
                } else {
                    proai.incrementAndGet();
                    ServiceExecuter exe = new ServiceExecuter(cmd);
                    pool.execute(exe);
                }
            }
            Logger.getLogger(ExchangerClient.class.getName()).log(Level.SEVERE, "ExchangerClient_{0}<->{1}_Stopped", new String[]{localIP(), remoteIP()});
        }
    }

    class ServiceExecuter implements Runnable {

        ExchangerCommand cmd;

        ServiceExecuter(ExchangerCommand cmd) {
            this.cmd = cmd;
        }

        @Override
        public void run() {
            Thread.currentThread().setName("Exchanger:" + cmd.toString());
            IService service = services.get(cmd.getServer());
            if (service == null) {
                sendError(ExchangerException.Type.SERVER_NOT_EXIST.create("Unknown Server " + cmd.getServer(), null), cmd);
            } else {
                try {
                    List<String> o;
                    try {
                        o = service.process(cmd.getArgsList());
                    } catch (ExchangerException ex) {
                        sendError(ex, cmd);
                        return;
                    } catch (Throwable ex) {
                        Logger.getLogger(Exchanger.class.getName()).log(Level.SEVERE, cmd.getNumber() + "_EXE_ERR", ex);
                        sendError(ExchangerException.Type.EXECUTE_ERR.create("EXE_RROR", ex), cmd);
                        return;
                    }
                    ExchangerCommand ans = cmd.replayAnswer(o);
                    ans.sendTo(ExchangerClient.this);
                } catch (IOException ex) {
                    Logger.getLogger(Exchanger.class.getName()).log(Level.WARNING, cmd.getNumber() + "_reply_IO_error", ex);
                }
            }
        }
    }
    //Orginal Command, Recived Command
    private final Map<String, Pair<ExchangerCommand, ExchangerCommand>> lockers = new ConcurrentHashMap<String, Pair<ExchangerCommand, ExchangerCommand>>();
    private final ConcurrentHashMap<String, IService> services = new ConcurrentHashMap<String, IService>();
    static AtomicInteger proai = new AtomicInteger();
    static AtomicInteger askai = new AtomicInteger();

    private void sendError(ExchangerException ex, ExchangerCommand cmd) {
        List<String> es = new LinkedList<String>();
        es.add(ex.getType().name());
        es.add(ex.getMessage());
        Throwable th = ex.getCause();
        while (th != null) {
            es.add(th.getClass().getSimpleName());
            es.add(th.getMessage());
            th = th.getCause();
        }
        ExchangerCommand command = cmd.replayError(es);
        try {
            command.sendTo(this);
        } catch (IOException ex1) {
            Logger.getLogger(Exchanger.class.getName()).log(Level.WARNING, cmd.getNumber() + "_reply_IO_error", ex1);
        }
    }
    //Executer Pool
    private ExecutorService pool;
    private Thread listener;

    public ExchangerClient(String ip, int port, ExecutorService pool) throws UnknownHostException, IOException {
        super(ip, port);
        this.pool = pool;
        listener = new Listener();
        listener.start();
    }

    public ExchangerClient(String ip, int port) throws UnknownHostException, IOException {
        this(ip, port, Executors.newSingleThreadExecutor());
    }
}
